package com.yj;

import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.springframework.cloud.stream.annotation.StreamListener;
import org.springframework.integration.annotation.ServiceActivator;
import org.springframework.integration.context.IntegrationContextUtils;
import org.springframework.messaging.handler.annotation.Headers;
import org.springframework.messaging.handler.annotation.Payload;
import org.springframework.messaging.support.ErrorMessage;
import org.springframework.stereotype.Component;

import java.util.Map;

@Component
@Slf4j
public class DemoConsumer {

    /**
     * 监听消息，指定binding名称，condition对消息进行过滤
     * @param message 消息对象
     */
    @StreamListener(value = MySink.DEMO_INPUT, condition = "headers['ROCKET_TAGS']=='yj'")
    public void onMessage(@Headers Map<String, Object> headers, @Payload DemoMessage message) {
        log.info("[onMessage][消息头信息：{}]", headers);
        log.info("[onMessage][线程编号:{} 消息内容：{}]", Thread.currentThread().getId(), message);
//        throw new RuntimeException("测试异常重试机制！");
    }

    /**
     * 指定通道的消费失败的处理，通道名为：<destination>.<group>.errors
     * @param errorMessage
     */
    @ServiceActivator(inputChannel = "DEMO-TOPIC-01.consumer-test-1.errors")
    public void handleError(ErrorMessage errorMessage) {
        log.error("[handleError][payload：{}]", ExceptionUtils.getRootCauseMessage(errorMessage.getPayload()));
        log.error("[handleError][headers：{}]", errorMessage.getHeaders());
    }

    /**
     * 全局监听消费失败的处理
     * @param errorMessage
     */
    @StreamListener(IntegrationContextUtils.ERROR_CHANNEL_BEAN_NAME) // errorChannel
    public void globalHandleError(ErrorMessage errorMessage) {
        log.error("[globalHandleError][payload：{}]", ExceptionUtils.getRootCauseMessage(errorMessage.getPayload()));
        log.error("[globalHandleError][headers：{}]", errorMessage.getHeaders());
    }
}